package com.atguigu.chapter05.Source;

import com.atguigu.chapter05.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;
import java.util.Random;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/3 14:19
 */
public class Flink04_MySource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // TODO Source - 自定义
       final MySource mySource = new MySource();

        env
                .addSource(mySource)
                .map(new MapFunction<WaterSensor, WaterSensor>() {
                    @Override
                    public WaterSensor map(WaterSensor value) throws Exception {
                        if (value.getVc() > 5){
                            mySource.cancel();
                        }
                        return value;
                    }
                })
                .print();


        env.execute();
    }


    public static class MySource implements SourceFunction<WaterSensor> {
        private static volatile boolean runningFlag = true;

        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            Random random = new Random();
            while (runningFlag) {
                ctx.collect(
                        new WaterSensor("sensor_" + random.nextInt(3),
                                System.currentTimeMillis(),
                                random.nextInt(10))
                );
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            runningFlag = false;
        }
    }
}
